package com.bigdata.flink.checkpoint;

import com.bigdata.flink.util.StreamEnvUtil;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.ExecutionCheckpointingOptions;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

// 上传jar yarn应用模式提交作业
// flink run-application -d -t yarn-application -c com.bigdata.flink.checkpoint.SavepointTest bigdata-flink-examples-1.0-SNAPSHOT.jar
// 可以通过stop 、cancel命令停止job 并指定保存点。【注意stop时source算子必须实现StopFunction接口 】
// run-application可以指定保存点、或者检查点路径启动，从而数据处理流程。但是不能更改状态后端
public class SavepointTest {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamEnvUtil.getEnv();

        env.enableCheckpointing(5000, CheckpointingMode.AT_LEAST_ONCE);
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();

        // checkpoint changelog模式，测试阶段, 默认状态后端 memory，stateTable间隔默认10分钟，changelog间断存到checkpoint
//     env.enableChangelogStateBackend(true);

        System.setProperty("HADOOP_USER_NAME", "root@123456");

        // 检查点配置
        // 1、设置检查点state存储位置
        checkpointConfig.setCheckpointStorage("hdfs://192.168.18.111:9000/chk");
        // 2、checkpoint超时时间，默认超过十分钟、则checkpoint失败
        checkpointConfig.setCheckpointTimeout(Duration.ofMinutes(10).toMillis());
        // 3、barria对其精准一次
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);
        // 4、设置并发checkpoint数量。默认1，如果存在多个说明背压严重 每次异步checkpoint十分耗时
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 5、最小的checkpoint间隔时间，设置了这个，那么就不会出现并发，上面的并发个数失效
        checkpointConfig.setMinPauseBetweenCheckpoints(1000);
        // 6、上面checkpoint最小时间间隔控制，checkpoint可以大于这个时间。下面是严格控制checkpoint触发周期
        checkpointConfig.setCheckpointInterval(5000);
        //7、作业状态JobStatus.CANCELED，是否保存最新一次成功的checkpoint备份数据
        // 主动cancel会删除checkpoint，如果是程序挂掉是无法删除的
        checkpointConfig.setExternalizedCheckpointCleanup(CheckpointConfig.ExternalizedCheckpointCleanup.DELETE_ON_CANCELLATION);
        //8、checkpoint容忍的失败次数，超过这个次数整个作业默认运行失败
        checkpointConfig.setTolerableCheckpointFailureNumber(2);


        // UC方式
        // 开启，必须设置成 EXACTLY_ONCE，这个精准一次是越过buffer之后，先来的barrier先备份数据，
        // 所有的barrier来了备份对应算子的数据同时上报该算子checkpoint状态
        // 并发不能大于1
        checkpointConfig.enableUnalignedCheckpoints();
        // 设置AC超时时间 大于0，则先启动AC，超时后启用UC。默认0，表示一开始直接使用UC模式 UC=un-align
        checkpointConfig.setAlignedCheckpointTimeout(Duration.ofSeconds(1));


        DataStreamSource<String> socketSource = StreamEnvUtil.getSocketSource(env);

        SingleOutputStreamOperator<Tuple2<String, Integer>> summed =
                socketSource.flatMap(new FlatMapFunction<String, Tuple2<String, Integer>>() {
                            @Override
                            public void flatMap(String value, Collector<Tuple2<String, Integer>> out) throws Exception {
                                String[] words = value.split(" ");
                                for (String word : words) {
                                    out.collect(new Tuple2<>(word, 1));
                                }
                            }
                        })
                        .returns(Types.TUPLE(Types.STRING, Types.INT))
                        .keyBy(f -> f.f0)
                        .sum(1);

        summed.print();

        env.execute();

    }
}
